package com.apps.sdses.flink141.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description mock水位信息写入MySQL
 * @date 2023/6/12 13:29
 */
public class Datagen2MySQL_WaterSensor {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 场景：mock水位信息到mysql
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id        BIGINT,\n" +     //采集信息id
                "    sid        STRING,\n" +    //水位传感器id
                "    ts   TIMESTAMP(3),\n" +    //数据采集时间
                "    flag   BOOLEAN,\n" +       //水位报警开关
                "    vc INT\n" +                //采集到的水位高度
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
                "  'number-of-rows' = '100', \n" +   //生成100条数据

                "  'fields.sid.length' = '8', \n" +   //传感器id长度

                "  'fields.id.kind' = 'sequence', \n" +
                "  'fields.id.start' = '1', \n" +
                "  'fields.id.end' = '1000', \n" +

                "  'fields.vc.min' = '10', \n" +
                "  'fields.vc.max' = '100'\n" +
                ")");


        //Sink
//        tableEnv.executeSql("CREATE TABLE sinkTable WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");

        tableEnv.executeSql("CREATE TABLE sinkTable WITH (\n" +
                "   'connector' = 'jdbc',\n" +
//                "   'url' = 'jdbc:mysql://localhost:3306/test',\n" +
                "   'url' = 'jdbc:mysql://192.168.102.155:33066/db_datacube_test',\n" +
                "   'table-name' = 'water_sensor',\n" +
                "   'username' = 'datacube_test',\n" +
                "   'password' = 'Gbd#XERA@19d1ZtD'\n" +
                ") LIKE sourceTable (EXCLUDING ALL)");



        //Insert
        tableEnv.executeSql("insert into sinkTable select * from sourceTable");





    }
}
